package org.fhm.zdte.common.service.message;

import org.fhm.substrate.annotation.Setup;
import org.fhm.substrate.manager.Bootstrap;
import org.fhm.zdte.common.ability.IGetIdentification;
import org.fhm.zdte.common.ability.IRequestTimeout;
import org.fhm.zdte.common.config.ProtocolConfiguration;
import org.fhm.zdte.common.constant.CommonConstant;
import org.fhm.zdte.common.constant.protocol.ContentType;
import org.fhm.zdte.common.constant.protocol.PacketType;
import org.fhm.zdte.common.constant.subcmd.RequestCmd;
import org.fhm.zdte.common.pojo.Address;
import org.fhm.zdte.common.pojo.Request;
import org.fhm.zdte.common.pool.ThreadPoolFactory;
import org.fhm.zdte.common.protocol.AbstractMessageStandard;
import org.fhm.zdte.common.protocol.AbstractMessageWrappers;
import org.fhm.zdte.common.standard.IChannelDispatcher;
import org.fhm.zdte.common.standard.IReceiveMessageManager;
import org.fhm.zdte.common.util.CommonUtil;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * The abstract implementation for {@link IReceiveMessageManager}.
 *
 * @author 谭波
 * @since 2023/11/16
 */
public abstract class AbstractReceiveMessageManager implements IReceiveMessageManager, IRequestTimeout {

    private static final Map<String, Set<String>> userMessagePool = new ConcurrentHashMap<>();

    private static final Map<String, byte[][]> messageBufferPool = new ConcurrentHashMap<>();

    private static final Map<String, ScheduledFuture<?>> scheduleMessageTaskPool = new ConcurrentHashMap<>();

    private static final Map<String, Long> timeoutRecordPool = new HashMap<>();
    private final ContentType contentType;
    private int lengthOfPacketWithContent;
    @Setup("->protocol.name")
    private AbstractMessageWrappers abstractMessageWrappers;

    @Setup
    private ThreadPoolFactory threadPoolFactory;

    @Setup
    private CommonUtil commonUtil;

    @Setup
    private IChannelDispatcher iChannelDispatcher;

    @Setup
    private IGetIdentification iGetIdentification;

    @Setup
    private ProtocolConfiguration protocolConfiguration;

    public AbstractReceiveMessageManager(ContentType contentType) {
        this.contentType = contentType;
    }

    @Override
    public void initial(AbstractMessageStandard message, Address address) {
        String messageNumber = message.getMessageNumber();
        iChannelDispatcher.obtainChannelManager(address)
                .sendMessage(abstractMessageWrappers.getAckMessageWithContent(
                        PacketType.REQUEST,
                        RequestCmd.INITIAL,
                        iGetIdentification.getIdentification(),
                        0,
                        0,
                        messageNumber,
                        null
                ).setContentType(message.getContentType()), address);
        if (Objects.isNull(messageBufferPool.get(messageNumber))
                && Objects.isNull(timeoutRecordPool.get(messageNumber))) {
            synchronized (this) {
                if (Objects.isNull(messageBufferPool.get(messageNumber))
                        && Objects.isNull(timeoutRecordPool.get(messageNumber))) {
                    messageBufferPool.put(messageNumber, new byte[message.getMessageLength()][]);
                    timeoutRecordPool.put(
                            messageNumber,
                            commonUtil.getCurTimeMillisSecond()
                    );
                    userMessagePool.computeIfAbsent(
                            message.getFrom(), k -> new CopyOnWriteArraySet<>()
                    ).add(messageNumber);
                    enableAnswerDetector(messageNumber, address);
                }
            }
        }
    }

    private void enableAnswerDetector(String messageNumber, Address address) {
        scheduleMessageTaskPool.put(
                messageNumber,
                threadPoolFactory.submitPeriodicTask(
                        () -> {
                            Long time;
                            if (
                                    Objects.isNull((time = timeoutRecordPool.get(messageNumber)))
                                            ||
                                            (time + 300) > commonUtil.getCurTimeMillisSecond()
                            ) {
                                return;
                            }
                            timeoutRecordPool.put(messageNumber, commonUtil.getCurTimeMillisSecond());
                            byte[][] bytes = messageBufferPool.get(messageNumber);
                            int messageLength = bytes.length;
                            List<byte[]> sendCollect = new ArrayList<>();
                            StringBuilder builder = new StringBuilder();
                            int contentLength = lengthOfPacketWithContent - 5;
                            for (int i = 0; i < messageLength; i++) {
                                if (Objects.nonNull(bytes[i])) {
                                    String separator = CommonConstant.PACKAGE_SEQUENCE_NUMBER_SEPARATOR.getName();
                                    builder.append(i).append(separator);
                                    int length = builder.length();
                                    if (builder.toString().getBytes(Bootstrap.charset)
                                            .length >= contentLength) {
                                        int endLen = String.valueOf(i).length() + separator.length();
                                        builder.delete(length - endLen, length);
                                        sendCollect.add(builder.toString().getBytes(Bootstrap.charset));
                                        builder = new StringBuilder();
                                        builder.append(i).append(separator);
                                    }
                                }
                            }
                            byte[] endContent = builder
                                    .toString()
                                    .getBytes(Bootstrap.charset);
                            int endContentLength = endContent.length;
                            if (endContentLength > 0 && endContentLength < contentLength)
                                sendCollect.add(endContent);
                            sendCollect.forEach(
                                    content ->
                                            iChannelDispatcher.obtainChannelManager(address)
                                                    .sendMessage(abstractMessageWrappers.getAckMessageWithContent(
                                                            PacketType.REQUEST,
                                                            RequestCmd.SEND,
                                                            iGetIdentification.getIdentification(),
                                                            messageLength,
                                                            0,
                                                            messageNumber,
                                                            content
                                                    ).setContentType(contentType), address)
                            );
                        },
                        100,
                        TimeUnit.MILLISECONDS
                )
        );
    }

    @Override
    public void addMessageToBuffer(
            AbstractMessageStandard messageStandard,
            Address address
    ) {
        String messageNumber = messageStandard.getMessageNumber();
        timeoutRecordPool.put(messageNumber, commonUtil.getCurTimeMillisSecond());
        int serialNumber = messageStandard.getSerialNumber();
        byte[][] bytes = messageBufferPool.get(messageNumber);
        if (Objects.nonNull(bytes))
            if (Objects.isNull(bytes[serialNumber])) {
                byte[] content = messageStandard.getContent();
                if (serialNumber == (messageStandard.getMessageLength() - 1)) {
                    bytes[serialNumber] = content;
                } else {
                    if (this.lengthOfPacketWithContent == content.length) {
                        bytes[serialNumber] = content;
                    }
                }
            }
    }

    @Override
    public Request isCompleteAndGet(String messageNumber) {
        byte[][] bytes = messageBufferPool.get(messageNumber);
        if (Objects.nonNull(bytes)) {
            synchronized (bytes) {
                if (Objects.nonNull(messageBufferPool.get(messageNumber))) {
                    if (Arrays.stream(bytes).allMatch(Objects::nonNull)) {
                        int size = bytes.length - 1;
                        int subArrLen = bytes[0].length;
                        int lastSize = bytes[size].length;
                        int totalLen = subArrLen * size + lastSize;
                        byte[] mergedArray = new byte[totalLen];
                        for (int i = 0, currentIndex = 0; i < size; i++, currentIndex += subArrLen) {
                            System.arraycopy(bytes[i], 0, mergedArray, currentIndex, subArrLen);
                        }
                        System.arraycopy(bytes[size], 0, mergedArray, (totalLen - lastSize), lastSize);
                        Request request = convert(mergedArray);
                        messageBufferPool.remove(messageNumber);
                        return request;
                    }
                }
            }
        }
        return null;
    }

    @Override
    public void sendCompleteScheduleAck(String messageNumber, Address address) {
        commonUtil.cancelTask(scheduleMessageTaskPool.remove(messageNumber));
        timeoutRecordPool.remove(messageNumber);
        scheduleMessageTaskPool.put(
                messageNumber,
                threadPoolFactory.submitPeriodicTask(
                        () -> iChannelDispatcher.obtainChannelManager(address)
                                .sendMessage(abstractMessageWrappers.getAckMessageWithContent(
                                        PacketType.REQUEST,
                                        RequestCmd.SEND,
                                        iGetIdentification.getIdentification(),
                                        0,
                                        0,
                                        messageNumber,
                                        CommonConstant
                                                .MESSAGE_COMPLETE_MARK
                                                .getName()
                                                .getBytes(Bootstrap.charset)
                                ).setContentType(contentType), address),
                        1,
                        TimeUnit.SECONDS
                )
        );
    }

    @Override
    public void cancelCompleteAck(String messageNumber) {
        commonUtil.cancelTask(scheduleMessageTaskPool.remove(messageNumber));
    }

    @Override
    public void clearMessageBuffer(String from) {
        Set<String> messageNumbers = userMessagePool.get(from);
        if (Objects.nonNull(messageNumbers)) {
            synchronized (messageNumbers) {
                if (Objects.nonNull(userMessagePool.get(from))) {
                    messageNumbers.forEach(
                            messageNumber -> {
                                messageBufferPool.remove(messageNumber);
                                timeoutRecordPool.remove(messageNumber);
                                commonUtil.cancelTask(scheduleMessageTaskPool.remove(messageNumber));
                            }
                    );
                    userMessagePool.remove(from);
                }
            }
        }
    }


    /**
     * Convert a byte array into the request object.
     *
     * @param buffer Byte array.
     * @return Request object.
     */
    public abstract Request convert(byte[] buffer);

    /**
     * Configure the length that Packet-Data with content.
     */
    protected void setLengthOfPacketWithContent() {
        lengthOfPacketWithContent = protocolConfiguration.getPacketLength()
                - abstractMessageWrappers.getLengthOfAckMessageWithContent(
                commonUtil.getIdentification(null)
        );
    }

    @Override
    public void connectTimeoutRun(String form, String messageNumber) {
        Set<String> messageNumbers = userMessagePool.get(form);
        if (Objects.nonNull(messageNumbers))
            userMessagePool.get(form).remove(messageNumber);
        commonUtil.cancelTask(scheduleMessageTaskPool.remove(messageNumber));
        timeoutRecordPool.remove(messageNumber);
        messageBufferPool.remove(messageNumber);
    }

}
